package com.siteseat.crudeq.broker;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.ChannelGroupFuture;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

import com.siteseat.common.foundation.PropertyHelper;
import com.siteseat.common.security.Credential;
import com.siteseat.common.security.SecureComponent;
import com.siteseat.common.security.impl.DefaultSecurityManager;
import com.siteseat.crudeq.broker.netty.NettySessionFactory;
import com.siteseat.crudeq.broker.netty.ServerPipelineFactory;
import com.siteseat.crudeq.broker.transport.BrokerSessionFactory;
import com.siteseat.crudeq.broker.transport.BrokerSessionManager;
import com.siteseat.crudeq.broker.transport.RequestHandlerRepository;
import com.siteseat.crudeq.broker.v1.request.V1RequestHandler;
import com.siteseat.crudeq.kernel.QueueManager;
import com.siteseat.crudeq.kernel.impl.QueueManagerImpl;

public class CrudeQBroker extends SecureComponent {

  private QueueManager queueManager;
  
  private ChannelFactory         channelFactory;

  private ServerBootstrap        bootstrap;

  private Map<String, Object>    options;

  private ServerPipelineFactory pipelineFactory;

  private Set<String>            addressToBind;

  private ChannelGroup           allChannels;

  private int                    tcpReceiveBufferSize = -1;

  private int                    tcpSendBufferSize    = -1;
  
  private int                    terminateTimeoutBySeconds = 10;

  /**
   * @param args
   * @throws Exception
   */
  public static void main( String[] args ) throws Exception {
    CrudeQBroker s = createDefault();
    s.init();
  }
  
  
  /**
   * Gets the terminateTimeoutBySeconds
   *
   * @return the terminateTimeoutBySeconds
   */
  public int getTerminateTimeoutBySeconds() {
    return terminateTimeoutBySeconds;
  }


  /**
   * Sets the terminateTimeoutBySeconds
   *
   * @param terminateTimeoutBySeconds the terminateTimeoutBySeconds to set
   */
  public void setTerminateTimeoutBySeconds( int terminateTimeoutBySeconds ) {
    this.terminateTimeoutBySeconds = terminateTimeoutBySeconds;
  }



  /**
   * Gets the queueManager
   *
   * @return the queueManager
   */
  public QueueManager getQueueManager() {
    return queueManager;
  }

  /**
   * Sets the queueManager
   *
   * @param queueManager the queueManager to set
   */
  public void setQueueManager( QueueManager queueManager ) {
    this.queueManager = queueManager;
  }

  /**
   * Gets the tcpReceiveBufferSize
   *
   * @return the tcpReceiveBufferSize
   */
  public int getTcpReceiveBufferSize() {
    return tcpReceiveBufferSize;
  }

  /**
   * Sets the tcpReceiveBufferSize
   *
   * @param tcpReceiveBufferSize the tcpReceiveBufferSize to set
   */
  public void setTcpReceiveBufferSize( int tcpReceiveBufferSize ) {
    this.tcpReceiveBufferSize = tcpReceiveBufferSize;
  }

  /**
   * Gets the tcpSendBufferSize
   *
   * @return the tcpSendBufferSize
   */
  public int getTcpSendBufferSize() {
    return tcpSendBufferSize;
  }

  /**
   * Sets the tcpSendBufferSize
   *
   * @param tcpSendBufferSize the tcpSendBufferSize to set
   */
  public void setTcpSendBufferSize( int tcpSendBufferSize ) {
    this.tcpSendBufferSize = tcpSendBufferSize;
  }

  /**
   * @return
   */
  public static CrudeQBroker createDefault() throws Exception {
    CrudeQBroker r = new CrudeQBroker();

    Set<String> addresses = new HashSet<String>();
    addresses.add( "localhost:8080" );
    r.setAddressToBind( addresses );

    Map<String, Object> opts = new HashMap<String, Object>();
    opts.put( "child.tcpNoDelay", Boolean.TRUE );
    opts.put( "child.keepAlive", Boolean.TRUE );
    opts.put( "child.reuseAddress", Boolean.TRUE );
    opts.put( "reuseAddress", Boolean.TRUE );

    if( r.tcpReceiveBufferSize != -1 ) {
      opts.put( "child.receiveBufferSize", Integer.valueOf( r.tcpReceiveBufferSize ) );
    }
    if( r.tcpSendBufferSize != -1 ) {
      opts.put( "child.sendBufferSize", Integer.valueOf( r.tcpSendBufferSize ) );
    }

    r.setOptions( opts );

    ChannelFactory cf = new NioServerSocketChannelFactory( Executors.newCachedThreadPool(), Executors
            .newCachedThreadPool(), Runtime.getRuntime().availableProcessors() * 2 );
    r.setChannelFactory( cf );

    BrokerSessionFactory sessionFactory = new NettySessionFactory();

    Credential admin = new Credential( "yqiang", true );
    admin.setPassword( "pwd" );
    admin.setEncryptionKey( "test".getBytes() );

    DefaultSecurityManager securityManager = new DefaultSecurityManager( admin );
    // securityManager.init();

    QueueManagerImpl queueManager = new QueueManagerImpl();
    queueManager.init();

    BrokerSessionManager sessionManager = new BrokerSessionManager( sessionFactory, queueManager );
    sessionManager.setSecurityManager( securityManager );
    sessionManager.init();

    RequestHandlerRepository requestFactoryRepository = new RequestHandlerRepository();
    V1RequestHandler requestHandler = new V1RequestHandler();
    requestHandler.init();
    requestFactoryRepository.register( requestHandler );

    ServerPipelineFactory pf = new ServerPipelineFactory( sessionManager, requestFactoryRepository );
    pf.init();
    r.setPipelineFactory( pf );

    return r;
  }

  /**
   * Gets the addressToBind
   * 
   * @return the addressToBind
   */
  public Set<String> getAddressToBind() {
    return addressToBind;
  }

  /**
   * Sets the addressToBind
   * 
   * @param addressToBind
   *          the addressToBind to set
   */
  public void setAddressToBind( Set<String> addressToBind ) {
    this.addressToBind = addressToBind;
  }

  /**
   * Gets the channelFactory
   * 
   * @return the channelFactory
   */
  public ChannelFactory getChannelFactory() {
    return channelFactory;
  }

  /**
   * Sets the channelFactory
   * 
   * @param channelFactory
   *          the channelFactory to set
   */
  public void setChannelFactory( ChannelFactory channelFactory ) {
    this.channelFactory = channelFactory;
  }

  /**
   * Gets the options
   * 
   * @return the options
   */
  public Map<String, Object> getOptions() {
    return options;
  }

  /**
   * Sets the options
   * 
   * @param options
   *          the options to set
   */
  public void setOptions( Map<String, Object> options ) {
    this.options = options;
  }

  /**
   * Gets the pipelineFactory
   * 
   * @return the pipelineFactory
   */
  public ServerPipelineFactory getPipelineFactory() {
    return pipelineFactory;
  }

  /**
   * Sets the pipelineFactory
   * 
   * @param pipelineFactory
   *          the pipelineFactory to set
   */
  public void setPipelineFactory( ServerPipelineFactory pipelineFactory ) {
    this.pipelineFactory = pipelineFactory;
  }

  /**
   * @return
   */
  protected ServerBootstrap createInternalBootstrap( ChannelFactory cf ) {
    return new ServerBootstrap( cf );
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void doInit() throws Exception {
    super.doInit();
    
    PropertyHelper.ensureInitialized( log, getQueueManager(), "queueManager" );

    ChannelFactory cf = getChannelFactory();
    PropertyHelper.ensureInitialized( log, cf, "channelFactory" );

    this.bootstrap = createInternalBootstrap( cf );

    ChannelPipelineFactory pf = getPipelineFactory();
    PropertyHelper.ensureInitialized( log, pf, "pipelineFactory" );
    this.bootstrap.setPipelineFactory( pf );

    Map<String, Object> opts = getOptions();
    if( opts != null ) {
      for( Map.Entry<String, Object> ent: opts.entrySet() ) {
        this.bootstrap.setOption( ent.getKey(), ent.getValue() );
      }
    }

    this.allChannels = new DefaultChannelGroup( "crude-broker" );

    Set<String> addrs = getAddressToBind();
    PropertyHelper.ensureHasElements( log, addrs, "addressToBind" );
    
    for( String addr: addrs ) {
      String[] hostAndPort = addr.split( ":" );
      if( hostAndPort.length != 2 ) {
        throw new Exception( "Invalid address form. The form should be <host>:<port>. For example: localhost:40577" );
      }
      InetAddress host = InetAddress.getByName( hostAndPort[0] );
      int port = Integer.parseInt( hostAndPort[1] );
      InetSocketAddress a = new InetSocketAddress( host, port ) ; 
      Channel ch = this.bootstrap.bind( a );
      log.info( "Bind to " + a );
      
      this.allChannels.add( ch );
    }
    
    getPipelineFactory().setEnabled( true );
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void doDestroy() throws Exception {
    ChannelGroupFuture future = this.allChannels.close();

    long closeStart = System.currentTimeMillis();
    while( future.awaitUninterruptibly( 1, TimeUnit.SECONDS ) == false ) {
      if( (System.currentTimeMillis() - closeStart) >= 1000 * getTerminateTimeoutBySeconds() ) {
        log.warn( "Timeout to wait for channels to close" );
        break;
      }
      log.info( "Waiting for channels to close" );
    }

    if( !future.isCompleteSuccess() ) {
      log.warn( "channel group did not completely close" );
      Iterator<Channel> iterator = future.getGroup().iterator();
      while( iterator.hasNext() ) {
        Channel ch = iterator.next();
        if( ch.isBound() ) {
          log.warn( ch + " is still connected to " + ch.getRemoteAddress() );
        }
      }
    }

    //getChannelFactory().releaseExternalResources();
  }

}
